package com.zyf.project.kvdb.client.core.impl;

import com.zyf.project.binaryproto.common.BinaryProtocol;
import com.zyf.project.kvdb.client.core.KVDBOperator;
import com.zyf.project.kvdb.protocol.config.ClientConfig;
import com.zyf.project.kvdb.protocol.constants.Constants;
import com.zyf.project.kvdb.protocol.exception.KVDBException;
import com.zyf.project.kvdb.protocol.exception.KVDBTimeoutException;
import com.zyf.project.kvdb.protocol.handler.impl.NettyClient;
import com.zyf.project.kvdb.protocol.proto.*;
import com.zyf.project.kvdb.protocol.proto.impl.KVDBMessage;
import com.zyf.project.kvdb.protocol.uri.KVDBURI;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.binary.BytesUtils;
import utils.binary.impl.Bytes;
import utils.common.StringUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * KVDB-SDK
 *
 * @author Zhou Yifan
 */
public class KVDBClient implements KVDBOperator {

    private static final Logger LOGGER = LoggerFactory.getLogger(KVDBClient.class);

    /**
     * 客户端参数配置
     */
    private ClientConfig config;
    /**
     * 保存当前所有连接，服务器地址加端口作为键值
     */
    private Map<String, NettyClient> clients = new HashMap<>();
    /**
     * 当前数据库实际操作者，当前选择数据库为单实例时使用{@link KVDBSingle}，否则{@link KVDBCluster}
     */
    private KVDBOperator operator;

    public KVDBClient(String url) throws KVDBException {
        this(new KVDBURI(url));
    }

    public KVDBClient(@NotNull KVDBURI uri) throws KVDBException {
        this(new ClientConfig(uri.getHost(), uri.getPort(), uri.getDatabase()));
    }

    public KVDBClient(ClientConfig config) throws KVDBException {
        this.config = config;
        start();
    }

    /**
     * 创建客户端等待就绪状态，当配置数据库不为空时执行切换数据库操作
     */
    private void start() throws KVDBException {
        clients.put(config.getHost() + config.getPort(), newNettyClient(config));
        if (!StringUtils.isEmpty(config.getDatabase())) {
            use(config.getDatabase());
        }
    }

    /**
     * <p>创建服务端连接，提供连接就绪回调接口</p>
     * <p>针对连接初始创建回调接口执行唤醒等待操作</p>
     * <p>针对服务掉线重连回调接口执行客户端重启操作</p>
     *
     * @param config 客户端参数配置
     * @return {@link NettyClient}
     * @throws KVDBException KVDB 可能的异常
     */
    private @NotNull NettyClient newNettyClient(ClientConfig config) throws KVDBException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        NettyClient client = new NettyClient(config, () -> {
            if (countDownLatch.getCount() > 0) {
                countDownLatch.countDown();
            } else {
                if (!StringUtils.isEmpty(config.getDatabase())) {
                    clients.get(config.getHost() + config.getPort()).send(KVDBMessage.use(config.getDatabase()));
                }
            }
        });
        try {
            boolean res = countDownLatch.await(config.getTimeout(), TimeUnit.MILLISECONDS);
            if (!res) {
                LOGGER.debug("newNettyClient : CountDownLatch await is failed!");
            }
            if (!client.isReady()) {
                throw new KVDBException("connect time out, make sure the kvdb server is started");
            }
        } catch (InterruptedException e) {
            LOGGER.error("wait interrupted", e);
            Thread.currentThread().interrupt();
            throw new KVDBException(e.getMessage());
        }
        return client;
    }

    /**
     * 切换数据库，获取数据库配置信息，自动切换数据库单例和集群操作模式
     *
     * @param db 数据库名
     * @return {@link DatabaseClusterInfo}
     * @throws KVDBException KVDB 可能的异常
     */
    public synchronized DatabaseClusterInfo use(String db) throws KVDBException {
        if (StringUtils.isEmpty(db)) {
            throw new KVDBException("database is empty");
        }
        Response response = clients.get(config.getHost() + config.getPort()).send(KVDBMessage.use(db));
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }
        try {
            DatabaseClusterInfo info = BinaryProtocol.decode(response.getResult()[0].toBytes(), DatabaseClusterInfo.class);
            config.setDatabase(db);
            if (info.isClusterMode()) {
                // 集群模式下，创建当前数据库所有服务节点连接，并执行切换数据库操作
                int n = info.getClusterItem().getURLs().length;
                NettyClient[] selectedClients = new NettyClient[n];
                for (int i = 0; i < n; i++) {
                    KVDBURI kvdburi = new KVDBURI(info.getClusterItem().getURLs()[i]);
                    NettyClient nettyClient;
                    if (!clients.containsKey(kvdburi.getHost() + kvdburi.getPort())) {
                        nettyClient = newNettyClient(new ClientConfig(kvdburi.getHost(), kvdburi.getPort(), kvdburi.getDatabase()));
                        clients.put(kvdburi.getHost() + kvdburi.getPort(), nettyClient);
                    } else {
                        nettyClient = clients.get(kvdburi.getHost() + kvdburi.getPort());
                    }
                    response = nettyClient.send(KVDBMessage.use(kvdburi.getDatabase()));
                    if (null == response) {
                        config.setDatabase(null);
                        throw new KVDBTimeoutException("time out");
                    } else if (response.getCode() == Constants.ERROR) {
                        config.setDatabase(null);
                        throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
                    }
                    selectedClients[i] = nettyClient;
                }
                operator = new KVDBCluster(db, selectedClients);
            } else {
                operator = new KVDBSingle(clients.get(config.getHost() + config.getPort()));
            }
            return info;
        } catch (IOException e) {
            LOGGER.error("use command error", e);
            String msg = e.getMessage();
            throw new KVDBException(!StringUtils.isEmpty(msg) ? msg : e.toString());
        }
    }

    /**
     * 创建数据库，仅支持本地连接管理服务端口进行操作
     *
     * @param databaseBaseInfo 数据库实例基础信息
     * @return 布尔值
     * @throws KVDBException KVDB 可能的异常
     */
    public boolean createDatabase(DatabaseBaseInfo databaseBaseInfo) throws KVDBException {
        Response response = clients.get(config.getHost() + config.getPort())
                .send(KVDBMessage.createDatabase(new Bytes(BinaryProtocol.encode(databaseBaseInfo, DatabaseBaseInfo.class))));
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }
        return true;
    }

    /**
     * 开启数据库实例，仅支持本地连接管理服务端口进行操作
     *
     * @param database 数据库名
     * @return 布尔值
     * @throws KVDBException KVDB 可能的异常
     */
    public boolean enableDatabase(String database) throws KVDBException {
        Response response = clients.get(config.getHost() + config.getPort())
                .send(KVDBMessage.enableDatabase(database));
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }
        return true;
    }

    /**
     * 关闭数据库实例，仅支持本地连接管理服务端口进行操作
     *
     * @param database 数据库名
     * @return 布尔值
     * @throws KVDBException KVDB 可能的异常
     */
    public boolean disableDatabase(String database) throws KVDBException {
        Response response = clients.get(config.getHost() + config.getPort())
                .send(KVDBMessage.disableDatabase(database));
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }

        return true;
    }

    /**
     * 删除数据库实例，仅支持本地连接管理服务端口进行操作
     *
     * @param database 数据库名
     * @return 布尔值
     * @throws KVDBException KVDB 可能的异常
     */
    public boolean dropDatabase(String database) throws KVDBException {
        Response response = clients.get(config.getHost() + config.getPort())
                .send(KVDBMessage.dropDatabase(database));
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }

        return true;
    }

    /**
     * 服务器集群配置，仅支持本地连接管理服务端口进行操作
     *
     * @return {@link ClusterItem} 数组
     * @throws KVDBException KVDB 可能的异常
     */
    public ClusterItem[] clusterInfo() throws KVDBException {
        Response response = clients.get(config.getHost() + config.getPort()).send(KVDBMessage.clusterInfo());
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }
        ClusterInfo clusterInfo = BinaryProtocol.decode(response.getResult()[0].toBytes(), ClusterInfo.class);
        return clusterInfo.getClusterItems();
    }

    /**
     * 当前服务器所有数据实例信息，仅支持本地连接管理服务端口进行操作
     *
     * @return {@link DatabaseBaseInfo} 数组
     * @throws KVDBException KVDB 可能的异常
     */
    public DatabaseBaseInfo[] showDatabases() throws KVDBException {
        Response response = clients.get(config.getHost() + config.getPort()).send(KVDBMessage.showDatabases());
        if (null == response) {
            throw new KVDBTimeoutException("time out");
        } else if (response.getCode() == Constants.ERROR) {
            throw new KVDBException(BytesUtils.toString(response.getResult()[0].toBytes()));
        }

        return BinaryProtocol.decode(response.getResult()[0].toBytes(), DatabaseBaseInfos.class).getBaseInfos();
    }

    /**
     * 存在性查询
     *
     * @param key 待查询的键
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean exists(Bytes key) throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.exists(key);
    }

    /**
     * 存在性查询，支持多个键
     *
     * @param keys 待查询的键
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean[] exists(Bytes... keys) throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        if (keys.length == 0) {
            throw new KVDBException("no key present");
        }
        return operator.exists(keys);
    }

    /**
     * 键值对获取
     *
     * @param key 键，{@link Bytes} 类型
     * @return 键对应的值，{@link Bytes} 类型
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public Bytes get(Bytes key) throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.get(key);
    }

    /**
     * 键值对获取
     *
     * @param keys 键，{@link Bytes} 类型数组
     * @return 键对应的值，{@link Bytes} 类型数组
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public Bytes[] get(Bytes... keys) throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        if (keys.length == 0) {
            throw new KVDBException("no key present");
        }
        return operator.get(keys);
    }

    /**
     * 设置键值对（添加）
     *
     * @param key   键
     * @param value 值
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean put(Bytes key, Bytes value) throws KVDBException {
        return put(key, value, false);
    }

    /**
     * 设置键值对（添加）
     *
     * @param key   键
     * @param value 值
     * @param async 是否异步（Promise 消息）
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean put(Bytes key, Bytes value, boolean async) throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.put(key, value, async);
    }

    /**
     * 开启批处理
     *
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean batchBegin() throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.batchBegin();
    }

    /**
     * 取消批处理
     *
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean batchAbort() throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.batchAbort();
    }

    /**
     * 提交批处理，服务器掉线重连后会丢失未提交的数据，未提交的 batch 对其他客户端连接而言是不可见的
     *
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean batchCommit() throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.batchCommit();
    }

    /**
     * 提交批处理
     *
     * @param size 此次批处理操作去重后的 key 数量
     * @return 布尔值
     * @throws KVDBException KVDB 可能存在的异常
     */
    @Override
    public boolean batchCommit(long size) throws KVDBException {
        if (StringUtils.isEmpty(config.getDatabase())) {
            throw new KVDBException("no database selected");
        }
        return operator.batchCommit(size);
    }

    /**
     * 关闭数据库
     */
    @Override
    public void close() {
        if (null != operator) {
            operator.close();
        }
        for (Map.Entry<String, NettyClient> entry : clients.entrySet()) {
            entry.getValue().stop();
        }
        clients.clear();
    }
}
